package com.ehotting.mqbd.core.service.mq;

import com.alibaba.fastjson.JSON;
import com.ehotting.mqbd.api.bean.MQDataAble;
import com.ehotting.mqbd.core.logger.LOGGER_KEY;
import com.ehotting.mqbd.core.logger.MQSendLogger;
import com.ehotting.mqbd.core.mq.MessageProducer;
import com.ehotting.mqbd.core.manager.ModuleManager;
import com.ehotting.mqbd.core.manager.MsgStateManager;
import com.ehotting.mqbd.core.manager.SolverStateManager;
import com.ehotting.mqbd.core.sync.MqSyncFutureMapHelper;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;

public abstract class MQBaseService implements MQServiceAble {

    protected MessageProducer messageProducer;

    protected MqSyncFutureMapHelper<MQDataAble> mqSyncFutureMapHelper;

    protected MQSendLogger logger;

    @Autowired
    protected SolverStateManager solverStateManager;
    @Autowired
    protected ModuleManager moduleManager;
    @Autowired
    protected MQDataHelper mqDataHelper;
    @Autowired
    protected MsgStateManager msgStateManager;

    public MQBaseService(MessageProducer messageProducer,
                         MqSyncFutureMapHelper<MQDataAble> mqSyncFutureMapHelper,
                         MQSendLogger logger) {
        this.messageProducer = messageProducer;
        this.mqSyncFutureMapHelper = mqSyncFutureMapHelper;
        this.logger = logger;
    }

    public void request(MQDataAble data){

        if(data==null){
            return;
        }
        mqDataHelper.setPreValue(data);
        try{
            messageProducer.send(data);
        }catch (Exception e){
            MDC.put(LOGGER_KEY.EXCEPTION,"request error");
            throw  e;
        }finally{
            logger.info();
        }
    }

    public MQDataAble request(MQDataAble data, boolean sync) {
        if(sync == false){
            request(data);
            return null;
        }
        return request( data, 10);
    }


    /**
     *
     * @param data 数据实体
     * @param timeout 超时，单位：秒
     * @return
     */
    public FutureTask<MQDataAble> requestFuture(MQDataAble data,long timeout){
        mqDataHelper.setPreValue(data);

        if(timeout > 0 ) {
            mqSyncFutureMapHelper.set(data.getSerial());
        }

        SendResult sendResult = messageProducer.send(data);

        if(sendResult!=null && SendStatus.SEND_OK == sendResult.getSendStatus() && timeout > 0 ) {
            Callable callable = new Callable<MQDataAble>() {
                @Override
                public MQDataAble call() {
                    MQDataAble result = null;
                    try {
                        result = mqSyncFutureMapHelper.get(data.getSerial(), timeout);
                        if(data.isLogResponse()){
                            MDC.put(LOGGER_KEY.RESPONSE, JSON.toJSONString(result));
                        }
                    } catch (InterruptedException e) {
                        MDC.put(LOGGER_KEY.EXCEPTION,"request time out("+timeout+"s)");
                        e.printStackTrace();
                    }finally{
                        logger.info();
                    }
                    return result;
                };
            };
            FutureTask<MQDataAble> futureTask = new FutureTask<MQDataAble>(callable);
            Thread thread = new Thread(futureTask);
            thread.start();
            return futureTask;
        }

        return null;
    }

    /**
     *
     * @param data 数据实体
     * @param timeout 超时，单位：秒
     * @return
     */
    public MQDataAble request(MQDataAble data,long timeout){
        mqDataHelper.setPreValue(data);
        MQDataAble msg = null;
        try {
            if(timeout > 0 ) {
                mqSyncFutureMapHelper.set(data.getSerial());
            }

            SendResult sendResult = messageProducer.send(data);

            if(sendResult!=null && SendStatus.SEND_OK == sendResult.getSendStatus() && timeout > 0 ) {
                msg = mqSyncFutureMapHelper.get(data.getSerial(), timeout);
                if(data.isLogResponse()){
                    MDC.put(LOGGER_KEY.RESPONSE, JSON.toJSONString(msg));
                }
            }

        } catch (InterruptedException e) {
            MDC.put(LOGGER_KEY.EXCEPTION,"request time out("+timeout+"s)");
            e.printStackTrace();
        }finally{
            logger.info();
        }
        return msg;
    }

    public void response(MQDataAble data){
        mqSyncFutureMapHelper.putAndNotify(data.getSerial(), data);
    }





}